/*
Package KcpServer
author: jamon
time: 2023-09-05
desc:
*/
package KcpServer

import (
	"bufio"
	"encoding/json"
	"errors"
	"fmt"
	"gitee.com/jamon/espoir_go/common"
	"gitee.com/jamon/espoir_go/logger"
	"github.com/xtaci/kcp-go"
	"io"
	"sync"
	"time"
)

type Manager struct {
	quitOnce sync.Once

	kcpServer *Server
	isEncrypt bool

	sessionExecuteController *common.SerializeController

	// 所有session相关的逻辑操作都在session主协程中执行
	sessionMap map[string]*SessionInfo
}

func NewManager() (mgr *Manager, err error) {
	mgr = &Manager{
		sessionExecuteController: common.NewSerializeController(MaxChannelNum),
		sessionMap:               make(map[string]*SessionInfo),
	}
	err = mgr.initKcpServer()
	if err != nil {
		return nil, err
	}
	return
}

func (mgr *Manager) initKcpServer() (err error) {
	kcpServer, err := Listen(":10000")
	if err != nil {
		err = fmt.Errorf("initKcpServer: %v", err)
		return
	}
	mgr.kcpServer = kcpServer
	return
}

func (mgr *Manager) Run() {

	// 启动会话管理、逻辑处理协程
	//go common.SafeRun(func() {
	//	mgr.sessionExecuteController.Run()
	//})

	go common.SafeRun(func() {
		mgr.RunTimeTask()
	})
	mgr.Listen()
}

func (mgr *Manager) Listen() {
	for {
		conn, e := mgr.kcpServer.Accept()
		if e != nil {
			if errors.Is(e, io.ErrClosedPipe) {
				logger.Debug("stop to accept kcp connections")
			} else {
				logger.Error("failed to accept kcp connection", logger.Kwargs{"err": e})
			}
			break
		}

		go common.SafeRun(func() {
			// 每个会话用一个单独的协程来接收消息
			kcpSession, ok := conn.(*kcp.UDPSession)
			if !ok {
				_ = conn.Close()
				return
			}
			sessionId := GetSessionId(kcpSession)
			info := &SessionInfo{
				SessionId: sessionId,
				Session:   kcpSession,
			}
			mgr.sessionExecuteController.Execute(func() {
				mgr.sessionMap[sessionId] = info
			}, "add session to session map")

			mgr.handleKcpSession(info)
		})
	}
}

func (mgr *Manager) Quit() {
	mgr.quitOnce.Do(func() {
		logger.Info("start quit kcp server")
		_ = mgr.kcpServer.Close()
		logger.Info("quit kcp server successfully")

		mgr.sessionExecuteController.Quit()
	})
}

func (mgr *Manager) handleKcpSession(info *SessionInfo) {
	defer func() {
		info.Close()
	}()
	reader := bufio.NewReader(info.Session) // 使用bufio.NewReader来逐行读取数据
	for {
		// 读取消息长度
		msgLenBytes := make([]byte, MsgHeaderLen)
		_, err := io.ReadFull(reader, msgLenBytes)
		if err != nil {
			if errors.Is(err, io.EOF) {
				logger.Debug("kcp session closed")
			} else {
				logger.Error("failed to read message length", logger.Kwargs{"err": err})
			}
			break
		}
		header := NewMsgHeader(msgLenBytes)

		// 读取消息内容
		msgData := make([]byte, header.MsgLen)
		_, err = io.ReadFull(reader, msgData)
		if err != nil {
			if errors.Is(err, io.EOF) {
				logger.Debug("kcp session closed")
			} else {
				logger.Error("failed to read message data", logger.Kwargs{"err": err})
			}
			break
		}

		plainData := mgr.getPlainData(msgData)
		mgr.sessionExecuteController.Execute(func() {
			mgr.handleRequestMsg(&MsgContext{
				msgData:   &MsgData{MsgId: header.MsgId, RawData: plainData},
				sessionId: info.SessionId,
			})
		}, "handle msg")

		// 处理消息
		logger.Debug("kcp session read", logger.Kwargs{"data": string(msgData)})
	}
}

func (mgr *Manager) getPlainData(msg []byte) []byte {
	if mgr.isEncrypt {
		return mgr.decrypt(msg)
	}
	return msg
}

func (mgr *Manager) getEncryptData(msg []byte) []byte {
	if mgr.isEncrypt {
		return mgr.encrypt(msg)
	}
	return msg
}

func (mgr *Manager) encrypt(msg []byte) []byte {
	return msg
}

func (mgr *Manager) decrypt(msg []byte) []byte {
	return msg
}

func (mgr *Manager) handleRequestMsg(msgCtx *MsgContext) {
	logger.Debug("receive msg", logger.Kwargs{"data": string(msgCtx.msgData.RawData),
		"msgId": msgCtx.msgData.MsgId})
	handler := GetMsgHandler(msgCtx.msgData.MsgId)
	if handler == nil {
		logger.Warn("no handler for msg", logger.Kwargs{"msgId": msgCtx.msgData.MsgId})
		return
	}

	ret := handler(msgCtx.msgData.RawData)
	mgr.sessionExecuteController.Execute(func() {
		mgr.handleReplyMsg(&MsgContext{
			msgData:   &MsgData{MsgId: msgCtx.msgData.MsgId, RawData: ret},
			sessionId: msgCtx.sessionId,
		})
	}, "handle reply msg")
}

func (mgr *Manager) handleReplyMsg(msgCtx *MsgContext) {
	logger.Debug("reply msg", logger.Kwargs{"msg": string(msgCtx.msgData.RawData),
		"msgId": msgCtx.msgData.MsgId})
	handler := GetMsgHandler(msgCtx.msgData.MsgId)
	if handler == nil {
		logger.Warn("no handler for msg", logger.Kwargs{"msgId": msgCtx.msgData.MsgId})
		return
	}

	msgDataBytes, err := json.Marshal(&msgCtx.msgData)
	if err != nil {
		logger.Error("failed to marshal msg", logger.Kwargs{"msg": msgCtx.msgData})
		return
	}
	sessionInfo, ok := mgr.sessionMap[msgCtx.sessionId]
	if !ok {
		logger.Warn("no session for msg", logger.Kwargs{"msgId": msgCtx.msgData.MsgId,
			"sessionId": msgCtx.sessionId})
		return
	}

	go common.SafeRun(func() {
		encryptData := mgr.getEncryptData(msgDataBytes)
		_, err := sessionInfo.Session.Write(encryptData)
		if err != nil {
			if errors.Is(err, io.EOF) {
				logger.Debug("kcp session closed")
			} else {
				logger.Error("failed to write msg", logger.Kwargs{"msg": msgCtx.msgData, "err": err})
				return
			}
		}
	})
}

func (mgr *Manager) RunTimeTask() {
	for {
		time.Sleep(1 * time.Second)
		mgr.sessionExecuteController.Execute(mgr.CheckSessionHeartbeat, "CheckSessionHeartbeat")
	}
}

func (mgr *Manager) CheckSessionHeartbeat() {
	for sessionId, info := range mgr.sessionMap {
		if info.IsHeartTimeout() {
			// 心跳超时, 断开连接
			info.Close()
			logger.Debug("close session", logger.Kwargs{"sessionId": sessionId})
		}
	}
}
